﻿using Confluent.Kafka;
using LJ.Abstractions.Config;
using Microsoft.Extensions.Logging;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace LJ.EventBus.FastDog.Transport.Kafka.Internal;

public class ConnectionPool : IConnectionPool
{
    private readonly ConcurrentQueue<IProducer<string, byte[]>> _producerPool = [];


    private readonly EventBusTransportConfig _eventBusTransportConfig;
    private int _maxSize;
    private int _pCount;

    public ConnectionPool(IConfigProvider configProvider, ILogger<ConnectionPool> logger)
    {
        _eventBusTransportConfig = configProvider.Get<EventBusTransportConfig>();

        _maxSize = _eventBusTransportConfig.ConnectionPoolSize;

        logger.LogDebug("Kafka servers: {servers}", _eventBusTransportConfig.Servers);
    }

    public IProducer<string, byte[]> RentProducer()
    {
        if(_producerPool.TryDequeue(out var producer))
        {
            Interlocked.Decrement(ref _pCount);

            return producer;
        }

        var config = new ProducerConfig(new Dictionary<string, string>(_eventBusTransportConfig.MainConfig))
        {
            BootstrapServers = _eventBusTransportConfig.Servers,
            QueueBufferingMaxMessages = 10,
            MessageTimeoutMs = 5000,
            RequestTimeoutMs = 3000
        };

        producer = new ProducerBuilder<string, byte[]>(config).Build();

        return producer;
    }

    public bool Return(IProducer<string, byte[]> producer)
    {
        if (Interlocked.Increment(ref _pCount) <= _maxSize)
        {
            _producerPool.Enqueue(producer);

            return true;
        }

        producer.Dispose();

        Interlocked.Decrement(ref _pCount);

        return false;
    }

    public void Dispose()
    {
        _maxSize = 0;

        while (_producerPool.TryDequeue(out var context))
        {
            context.Dispose();
        }
        GC.SuppressFinalize(this);
    }

    
}
